package edu.kit.aifb.ldstream.repository.listeners;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.util.LinkedList;
import java.util.List;

import org.apache.http.HttpEntity;
import org.apache.http.HttpResponse;
import org.apache.http.client.ClientProtocolException;
import org.apache.http.client.HttpClient;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.DefaultHttpClient;

import edu.kit.aifb.ldstream.env.Environment;
import edu.kit.aifb.ldstream.repository.event.EventManager;

/**
 * Listening from the given source and saving it to the global repository.
 * 
 * @author Holger
 *
 */

public class HTTPListeningThread extends AbstractListeningThread{

	private List<Double> jitterList = new LinkedList<Double>();
	private List<Integer> bandwidthList = new LinkedList<Integer>();

	private String stream;
	private int listenTime;
	private EventManager eventManager;
	private int numEvents = 0;
	
	/**
	 * 
	 * @param stream URL to be listened.
	 * @param listenTime in milliseconds.
	 */
	public HTTPListeningThread(String stream, String baseURI, int listenTime, EventManager manager){
		super(baseURI);
		this.listenTime = listenTime;
		this.stream = stream;
		this.eventManager = manager;
	}

	public void run() {


		HttpClient httpclient = new DefaultHttpClient();
		HttpGet httpget = new HttpGet(this.stream);

		try {
			// convert HttpResponse to BufferedReader
			HttpResponse response = httpclient.execute(httpget);
			HttpEntity entity = response.getEntity();
			InputStream instream = entity.getContent();
			BufferedReader reader = new BufferedReader(new InputStreamReader(instream));

			// waiting for Stream
			int waitAttempts = Environment.STREAM_WAITING_TIME / Environment.SLEEP_TIME;
			while(reader.ready() == false && waitAttempts > 0){
				try {
					Thread.sleep(Environment.SLEEP_TIME);
					waitAttempts--;
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}

			// reading the stream and append it to a string
			if(Environment.VERBOSE){
				System.out.println("Starting to listen: " + this.stream);
			}
			this.numEvents = 0;
			int sumFrequences = 0;
			double actualJitter;
			long endofEvent;
			// initial capacity
			if(reader.ready()){
				StringBuilder sb = new StringBuilder(1024);
				boolean startOfEvent = true;
				long start = System.currentTimeMillis();
				long now = start;
				int numTriples = 0;
				endofEvent = start;
				String s = reader.readLine();

				while(now - start <= this.listenTime && s != null){
					now = System.currentTimeMillis();
					numTriples++; // every line is a triple
					if(startOfEvent){
						// at the beginning we accept the @prefixes
						if(s.contains("@prefix")){
							sb.append(s);
							sb.append("\n");
						}else{
							sb.append(s);
							sb.append("\n");
							startOfEvent = false;
						}
					}else{
						// if we find @prefix here we have a new event
						if(!s.contains("@prefix")){
							sb.append(s);
							sb.append("\n");
						}else{
							// start of new event found
							// send the string to the repository and continue listening
							sumFrequences += now - endofEvent;
							// calculating the jitter by calculating the difference between the 
							// mean event time and the duration of the last event. 
							this.numEvents++;
							actualJitter = Math.abs((((double)sumFrequences) / this.numEvents) - (now - endofEvent));
							// adding the jitter for calculating jitter distribution later on
							if(actualJitter != 0){
								this.jitterList.add(actualJitter);
							}
							// adding the number of triples of the event for calculating the distribution of eventsize
							this.bandwidthList.add(numTriples);
							// resetting the event size
							numTriples = 0;

							startOfEvent = true;
							this.eventManager.addEvent(sb.toString(), this.baseURI, this.stream);
							sb = new StringBuilder(sb.length());
							sb.append(s);
							endofEvent = System.currentTimeMillis();
							sb.append("\n");
						}
					}
					s = reader.readLine();
				}

				// try to save the rest of the event after abort
				this.eventManager.addEvent(sb.toString(), this.baseURI, this.stream);
				if(Environment.VERBOSE){
					System.out.println("Successfully listened to " + this.stream);
				}

			}

		} catch (ClientProtocolException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		}

	}
	@Override
	public int getNumEvents(){
		return this.numEvents;
	}
	
	/**
	 * @return the jitterList
	 */
	public List<Double> getJitterList() {
		return jitterList;
	}

	/**
	 * @return the bandwidthList
	 */
	public List<Integer> getBandwidthList() {
		return bandwidthList;
	}

	/**
	 * @return the stream
	 */
	public String getStream() {
		return stream;
	}

	/**
	 * @return the listenTime
	 */
	public int getListenTime() {
		return listenTime;
	}

	/**
	 * @return the eventManager
	 */
	public EventManager getEventManager() {
		return eventManager;
	}

}
